package club.drguo.hadoop.mapreduce.enhance;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import club.drguo.hadoop.mapreduce.topkurl.TopKeyURLRunner;

public class EnhanceRunner  extends Configured implements Tool{
	@Override
	public int run(String[] args) throws Exception {
		Configuration configuration = new Configuration();
		Job job = Job.getInstance(configuration);
		job.setJarByClass(EnhanceRunner.class);
		
		job.setMapperClass(LogEnhanceMapper.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(NullWritable.class);
		
		job.setOutputFormatClass(EnhanceOutputFormat.class);
		
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		//虽然EnhanceOutputFormat已经有了，但下面必须写，可以产生success
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		return job.waitForCompletion(true)?0:1;
	}
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new TopKeyURLRunner(), args);
		System.exit(res);
	}
	
}
